package com.huak.zookeeper.distributedlock.util;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;

import java.util.concurrent.CountDownLatch;

/**
 * 分布式锁的工具类
 * @author liuhuan
 * @date 2018/11/19 15:54
 * @description
 */
@Slf4j
public class DistributedLock {

    //zk客户端
    private CuratorFramework client=null;

    //用于挂起当前请求，并且等待上一个分布式锁的释放
    private static CountDownLatch zkLockLatch=new CountDownLatch(1);

    //分布式总节点名称
    private static final String ZK_LOCK_PROJECT="imooc-locks";

    private static final String DISTRIBUTED_LOCK="distributed_lock";

    public DistributedLock(CuratorFramework client){
        this.client=client;
    }

    /**
     * 初始化锁
     */
    public void init(){
        //设置命名空间
        client=client.usingNamespace("ZKLocks-NameSpace");

        try{
            if(client.checkExists().forPath("/"+ZK_LOCK_PROJECT)==null){
                client.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath("/"+ZK_LOCK_PROJECT);

            }
            //针对zk的分布式节点，创建相应的watcher事件监听
            addWatchToLock("/"+ZK_LOCK_PROJECT);
        }catch (Exception e){
            log.error(e.getMessage(),e);
        }

    }

    public void addWatchToLock(String path)  throws Exception{
        final PathChildrenCache cache=new PathChildrenCache(client,path,true);
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        cache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)){
                    String path=event.getData().getPath();
                    log.info("上一个会话以释放锁或者会话结束，节点路径为{}",path);
                    //
                    if(path.contains(DISTRIBUTED_LOCK)){
                        log.info("释放计数器，让当前请求来获取分布式锁");
                        zkLockLatch.countDown();
                    }
                }
            }
        });

    }

    //获取锁
    public void getLock(){
        while (true){
            try {
                client.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.EPHEMERAL)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath("/"+ZK_LOCK_PROJECT+"/"+DISTRIBUTED_LOCK);
                log.info("获取分布式锁成功。。。");
                return;
            }catch (Exception e){
                log.info("获取分布式锁失败。。。");
                //如果没有获取到锁
                if(zkLockLatch.getCount()<=0){
                    zkLockLatch=new CountDownLatch(1);
                }
                try {
                    //阻塞线程
                    zkLockLatch.await();
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }
        }
    }

    /**
     * 释放分布式锁
     * @return
     */
    public boolean releaseLock(){
        try{
            if(client.checkExists().forPath("/"+ZK_LOCK_PROJECT+"/"+DISTRIBUTED_LOCK)!=null){
                client.delete().forPath("/"+ZK_LOCK_PROJECT+"/"+DISTRIBUTED_LOCK);
            }
        }catch (Exception e){
            log.error(e.getMessage(),e);
            return false;
        }
        return true;
    }

}
